package com.hacker.spider.concurrent;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

import com.hacker.spider.Spider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ResultActuator {

  public static final LongAdder ADD_TASK_COUNT = new LongAdder();

  public static final LongAdder ADD_RETRY_TASK_COUNT = new LongAdder();

  private static final int EXE_QUEUE_SIZE = 1024;

  private static final Logger logger = LoggerFactory.getLogger(Spider.class);

  private static final ResultActuator RESULT_ACTUATOR = new ResultActuator();

  private final ThreadPoolExecutor executor;

  private final ThreadPoolExecutor rejectedTaskRetryExecutor;

  private ResultActuator() {
    executor = new ThreadPoolExecutor(32, 48, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(EXE_QUEUE_SIZE),
      new HackerThreadFactory("pool-"), new HackerRetrySubmitRejectedExecutionHandler());
    rejectedTaskRetryExecutor = new ThreadPoolExecutor(5, 5, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(),
      new HackerThreadFactory("rejectedTaskRetry-"), new HackerRetrySubmitRejectedExecutionHandler());
  }

  public static synchronized ResultActuator getInstance() {
    return RESULT_ACTUATOR;
  }

  public void addTask(Runnable task) {
    ADD_TASK_COUNT.increment();
    try {
      executor.submit(task);
    } catch (Exception e) {
      logger.error("addTask error.", e);
      ADD_TASK_COUNT.decrement();
    }
  }

  public boolean isFinish() {
    return (ADD_TASK_COUNT.longValue() == executor.getCompletedTaskCount()) && (executor.getQueue().size() == 0) && (
      ADD_RETRY_TASK_COUNT.longValue() == rejectedTaskRetryExecutor.getCompletedTaskCount()) && (
      rejectedTaskRetryExecutor.getQueue().size() == 0);
  }

  public void shutDown() {
    executor.shutdown();
    rejectedTaskRetryExecutor.shutdown();
  }

  public void rejectRetry(Runnable r) {
    try {
      ADD_RETRY_TASK_COUNT.increment();
      rejectedTaskRetryExecutor.submit(() -> {
        try {
          try {
            if (executor.getQueue().size() == EXE_QUEUE_SIZE) {
              // 粗略判断当队列不满时不等直接放进去.
              Thread.sleep(1000);
            }
          } catch (Exception e) {
            logger.error("rejectRetry sleep fail.", e);
          }
          addTask(r);
        } catch (Throwable e) {
          ADD_RETRY_TASK_COUNT.decrement();
          rejectRetry(r);
        }
      });
    } catch (Exception e) {
      ADD_RETRY_TASK_COUNT.decrement();
      logger.error("rejectRetry error.", e);
    }
  }

  private static class HackerThreadFactory implements ThreadFactory {

    private static final LongAdder POOL_NUMBER = new LongAdder();

    private final ThreadGroup group;

    private final AtomicInteger threadNumber = new AtomicInteger(1);

    private final String namePrefix;

    HackerThreadFactory(String prefix) {
      SecurityManager s = System.getSecurityManager();
      group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
      POOL_NUMBER.increment();
      namePrefix = prefix + POOL_NUMBER.longValue() + "-thread-";
    }

    @Override public Thread newThread(Runnable runnable) {
      Thread t = new Thread(group, runnable, namePrefix + threadNumber.getAndIncrement(), 0);
      if (t.isDaemon()) {
        t.setDaemon(false);
      }
      if (t.getPriority() != Thread.NORM_PRIORITY) {
        t.setPriority(Thread.NORM_PRIORITY);
      }
      return t;
    }
  }

  /**
   * 重试提交直到成功
   */
  private static class HackerRetrySubmitRejectedExecutionHandler implements RejectedExecutionHandler {

    @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
      ADD_TASK_COUNT.decrement();
      ResultActuator.getInstance().rejectRetry(r);
    }
  }
}
